import json
import asyncio
import asyncio_mqtt
import multiprocessing
import AsyncMQTTClientHandel

from utils import settings
from utils.util import Util
from utils.async_kafka_driver import AsyncKafkaProduce
from utils.async_minio_api import AsyncMinioUtil
from utils.codes import code_dict, prefixes
from utils.restful_tdengine import create_series_stables


def split_group(group_num=5):
    codes = ['Doton/Online/81000001/#', ]
    topics = [f'Doton/Online/80000{i}' for i in range(1, 9)]
    group_list = [codes[i::group_num] for i in range(group_num)]
    group_list = [gl for gl in group_list if gl]
    return group_list


def start_prefix(string):
    """判断测点名称是否以 temp、rms 开头"""
    return any(string.startswith(prefix) for prefix in prefixes)


def is_stables():
    """检测超级表是否存在"""
    logger = Util.pyfile_logger(__file__)
    print('正在检测超级表...')
    try:
        # 获取所有温度、有效值测点
        points = [uid for lines in code_dict.values() for uid in lines[1].values() if start_prefix(uid)]
        # 判断是否存在超级表，不存在则创建超级表
        create_series_stables(points)
        print('超级表检测完成')
    except Exception as e:
        print(f'超级表创建失败... {e}')
        logger.error(e)


async def mqtt_connect_handle(topics, kafka_producer, minio_producer):
    """
    协程主函数
    连接MQTT：处理接受信息
    topic: Doton/Online/81000001/IndexData/3526
    """
    # 可选：限制最大并发任务数（例如最多同时处理100个消息）
    # semaphore = asyncio.Semaphore(100)
    try:
        async with asyncio_mqtt.Client(
            hostname=settings.MQTT_SETTING['hostname'],
            port=settings.MQTT_SETTING['port'],
            username=settings.MQTT_SETTING['username'],
            password=settings.MQTT_SETTING['password']
        ) as client:
            await client.subscribe(topic=','.join(topics))
            async with client.messages() as message:
                async for msg in message:
                    topic_name = msg.topic.value
                    topicstrs = topic_name.split('/')
                    # 创建并调度任务
                    asyncio.create_task(
                        AsyncMQTTClientHandel.handel(
                            topicstrs, msg.payload, kafka_producer, minio_producer
                        )
                    )
    except asyncio_mqtt.MqttError as e:
        print(f"MQTT 错误: {e}")
        await asyncio.sleep(5)  # 等待后重连
        await mqtt_connect_handle(topics, kafka_producer, minio_producer)


def run(topics):
    # 进程内初始化 Kafka
    kafka_producer = AsyncKafkaProduce(settings.KAFKA_ALGO_PUSH_SETTING["kafka_servers"])
    # 初始化 Minio
    minio_producer = AsyncMinioUtil()
    asyncio.run(mqtt_connect_handle(topics, kafka_producer, minio_producer))


if __name__ == '__main__':
    # 检测是否创建了测点超级表
    is_stables()
    # 数据转存启动
    print('数据转存启动...')
    multiprocessing.set_start_method("spawn")
    group_list = split_group()
    # run(group_list[0])
    processes = []
    for topics in group_list:
        p = multiprocessing.Process(
            target=run,
            args=(topics,),
            name=f'Process-{group_list.index(topics)}'
        )
        processes.append(p)
        p.start()

    for pro in processes:
        pro.join()
